package com.bdi.mqtt_broker.handler;

import com.bdi.mqtt_broker.config.BrokerConfig;
import com.bdi.mqtt_broker.config.BrokerContextConfig;
import com.bdi.mqtt_broker.constant.Constants;
import com.bdi.mqtt_broker.model.InternalMessage;
import com.bdi.mqtt_broker.mqtt_protocol.ProtocolProcess;
import com.bdi.mqtt_broker.service.ChannelManagerService;
import com.bdi.mqtt_broker.service.ISessionStoreService;
import com.bdi.mqtt_broker.service.ISubscribeStoreService;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @author WangNing yogehaoren@gmial.com <br>
 * @since 1.0
 * MQTT 消息处理器
 * 目前只支持 Qos = 0, 支持消息订阅
 * 不支持  Qos = 1; Qos = 2
 * 不支持  遗嘱消息, Retain 消息
 * 不支持  客户端认证
 * 支持全部 MQTT 协议的实现参考 https://github.com/Wizzercn/MqttWk
 *
 */
@Component
@ChannelHandler.Sharable
public class BrokerHandler  extends SimpleChannelInboundHandler<MqttMessage> {

    @Autowired
    BrokerConfig brokerConfig;

    /**
     * ChannelGroup 存储Channel 保证以后可以通过Channel发送 MQTT Publish 消息
     */
    @Autowired
    private  ChannelGroup channelGroup;

    @Autowired
    private ChannelManagerService channelManagerService;

    @Autowired
    ProtocolProcess protocolProcess;

    @Autowired
    ISessionStoreService sessionStoreService;

    @Autowired
    ISubscribeStoreService subscribeStoreService;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.channelGroup.add(ctx.channel());
        this.channelManagerService.put(brokerConfig.getBrokerId() + "_" + ctx.channel().id().asLongText(), ctx.channel().id());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        this.channelGroup.remove(ctx.channel());
        this.channelManagerService.remove(brokerConfig.getBrokerId()  + "_" + ctx.channel().id().asLongText());
    }




    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {

        InternalMessage internalMessage = null;

        if (msg.decoderResult().isFailure()) {
            // 处理版本不兼容以及错误的连接标识
            Throwable cause = msg.decoderResult().cause();
            if (cause instanceof MqttUnacceptableProtocolVersionException) {
                ctx.writeAndFlush(MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION, false),
                        null));
            } else if (cause instanceof MqttIdentifierRejectedException) {
                ctx.writeAndFlush(MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false),
                        null));
            }
            ctx.close();

            return;
        }

        switch (msg.fixedHeader().messageType()) {
            case CONNECT:
                protocolProcess.connect().processConnect(ctx.channel(), (MqttConnectMessage) msg);
                break;
            case DISCONNECT:
                protocolProcess.disConnect().processDisConnect(ctx.channel(), msg);
                break;
            case PINGREQ:
                protocolProcess.pingReq().processPingReq(ctx.channel(), msg);
                break;
            case PINGRESP:
                break;
            case SUBSCRIBE:
                protocolProcess.subscribe().processSubscribe(ctx.channel(), (MqttSubscribeMessage) msg);
                break;
            case UNSUBSCRIBE:
                protocolProcess.unSubscribe().processUnSubscribe(ctx.channel(), (MqttUnsubscribeMessage) msg);
                break;
            case PUBLISH:


                MqttPublishMessage newMessage = (MqttPublishMessage)msg;
                byte[] messageBytes  = new byte[newMessage.payload().readableBytes()];
                newMessage.payload().getBytes(newMessage.payload().readerIndex(), messageBytes);
                protocolProcess.publish().processPublish(ctx.channel(), newMessage);

                internalMessage = new InternalMessage();
                internalMessage.setBrokerId(brokerConfig.getBrokerId());
                internalMessage.setClientId((String) ctx.channel().attr(AttributeKey.valueOf(Constants.CHANNEL_ATTRIBUTE_CLIENT_ID)).get());
                internalMessage.setMessageBytes(messageBytes);
                internalMessage.setTopic(newMessage.variableHeader().topicName());
                internalMessage.setMqttQoS(msg.fixedHeader().qosLevel().value());

                break;
            default:
                break;
        }

        if(internalMessage!=null){
            ctx.fireChannelRead(internalMessage);
        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (cause instanceof IOException) {

            // 远程主机强迫关闭了一个现有的连接的异常
            ctx.close();
            this.channelGroup.remove(ctx.channel());
            this.channelManagerService.remove(brokerConfig.getBrokerId()  + "_" + ctx.channel().id().asLongText());
            this.sessionStoreService.remove((String) ctx.channel().attr(AttributeKey.valueOf(Constants.CHANNEL_ATTRIBUTE_CLIENT_ID)).get());
            // TODO: MQTT客户端异常断线时，是否要清除该客户端之前的订阅信息？
            //   this.subscribeStoreService.removeForClient((String) ctx.channel().attr(AttributeKey.valueOf(Constants.CHANNEL_ATTRIBUTE_CLIENT_ID)).get());

        } else {
            super.exceptionCaught(ctx, cause);
        }

    }
}
